//package com.xiaomaoguai;
//
//import com.zhongan.blcs.client.DBEvent;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.flink.api.common.restartstrategy.RestartStrategies;
//import org.apache.flink.streaming.api.CheckpointingMode;
//import org.apache.flink.streaming.api.TimeCharacteristic;
//import org.apache.flink.streaming.api.environment.CheckpointConfig;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
//import org.apache.kafka.clients.consumer.ConsumerConfig;
//import org.eclipse.aether.util.ConfigUtils;
//
//import java.util.Properties;
//
//
//@Slf4j
//public class XmagicTagProcessJob {
//
//	private static final String SINK_NAME = "RedisSink";
//
//	private static final String JOB_NAME = "keplerTagProccess";
//
//	private static final String GROUP_NAME = "XmagicTagProccess";
//
//	private static final String DESERIALIZER_STRATEGY = "org.apache.kafka.common.serialization.StringDeserializer";
//
//	private static final String SOURCE_NAME = "kafkaSource";
//
//	private static final String CONSUMER_STRATEGY = "latest";
//
//	private static final String REDIS_HOST = "10.253.10.152:27379,10.253.9.90:27379,10.253.101.135:27379,10.253.8.14:27379,10.139.102.143:27379";
//
//	private static final String PREFIX = "FcpXmagic";
//
//	private static final String FORMAT = "yyyy-MM-dd";
//
//
//	public static void main(String[] args) throws Exception {
//		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//		//checkpointing时间间隔为1分钟
//		env.enableCheckpointing(60 * 1000);
//		env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//		env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//		// 重启次数 + 重启时间间隔
//		env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, org.apache.flink.api.common.time.Time.seconds(10L)));
//		//window的时间采用事件时间
//		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//
//		Properties props = new Properties();
//		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, ConfigUtils.getKafkaHost());
//		props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_NAME);
//		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, DESERIALIZER_STRATEGY);
//		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, DESERIALIZER_STRATEGY);
//		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, CONSUMER_STRATEGY);
//
//		//设置输入source
//		FlinkKafkaConsumer011<DBEvent> kafkaConsumerSource = new FlinkKafkaConsumer011<DBEvent>("", new DecisionApplyDtoDeserializationSchema(), props);
//
//		//类型转换
//		env.addSource(kafkaConsumerSource);
//
//		env.execute(JOB_NAME);
//		log.info("flink job={} start", JOB_NAME);
//	}
//
//}
